package cluster

import(
	"github.com/sniperHW/kendynet"
	"github.com/sniperHW/kendynet/rpc"
	"sanguo/codec/ss"
	"sanguo/common"
	"time"
	"fmt"
)


func onEstablishClient(end *endPoint,session kendynet.StreamSession) {

	//再次确保endPoint是有效的
	if nil == getEndPointByID(end.toPeerID()) {
		Infof("dialOK %s but endPoint leave cluster\n",end.toPeerID().ToString())
		dialError(end,session,fmt.Errorf("invaild peer"))
		return		
	}

	if nil != end.conn {
		Infof("endPoint:%s already have conn\n",end.toPeerID().ToString())
		session.Close("invaild session",0)
		return
	}
	onEstablish(end,session,true)

}

func onEstablishServer(service Service,session kendynet.StreamSession) bool {

	if service.ToPeerID() != selfService.ToPeerID() {
		end := getEndPointByID(service.ToPeerID())
		if nil == end {
			return false
		}

		if end.conn != nil {
			return false
		}

		onEstablish(end,session,false)

	} else {

		//自连接不需要建立endpoint
		rpcchan := newRPCChannel(session,service.ToPeerID().ToString())

		session.SetRecvTimeout(common.HeartBeat_Timeout * time.Second)
		session.SetReceiver(ss.NewReceiver("ss","rpc_req","rpc_resp"))
		session.SetEncoder(ss.NewEncoder("ss","rpc_req","rpc_resp"))
		session.Start(func (event *kendynet.Event) {
			if event.EventType == kendynet.EventTypeError {
				event.Session.Close(event.Data.(error).Error(),0)
			} else {
				switch event.Data.(type) {
					case *ss.Message:
						//处理普通消息
						queue.PostNoWait(func(){
								dispatchServer(session,event.Data.(*ss.Message))
						})
						break
					case *rpc.RPCRequest:
						queue.PostNoWait(func() {
							rpcServer.OnRPCMessage(rpcchan,event.Data.(*rpc.RPCRequest))
						})
						break	
					default:
						Errorf("invaild message type\n")
						break
				}
			}
		})
	}

	return true
}

func onEstablish(end *endPoint,session kendynet.StreamSession,isClient bool) {

	rpcchan := newRPCChannel(session,end.toPeerID().ToString())

	conn := &connection{
		session:session,
	}

	conn.rpcCli , _ = rpc.NewClient(rpcchan,&decoder{},&encoder{})
	
	if isClient {
		nextHeartBeat := time.Now().Unix() + (common.HeartBeat_Timeout/2)
		conn.nextHeartbeat = &nextHeartBeat
	}

	session.SetRecvTimeout(common.HeartBeat_Timeout * time.Second)
	session.SetReceiver(ss.NewReceiver("ss","rpc_req","rpc_resp"))
	session.SetEncoder(ss.NewEncoder("ss","rpc_req","rpc_resp"))
	session.SetCloseCallBack(func (sess kendynet.StreamSession, reason string) {
		queue.PostNoWait(func() {
			err := fmt.Errorf(reason)
			conn.rpcCli.OnChannelClose(err)
			end.conn = nil
			remSessionPeerID(session)
			Infof("%s disconnected error:%s\n",end.toPeerID().ToString(),reason)
		})
	})
	end.conn = conn

	session.Start(func (event *kendynet.Event) {
		if event.EventType == kendynet.EventTypeError {
			event.Session.Close(event.Data.(error).Error(),0)
		} else {
			switch event.Data.(type) {
				case *ss.Message:
					//处理普通消息
					queue.PostNoWait(func(){
						if isClient {
							dispatchClient(session,event.Data.(*ss.Message))
						} else {
							dispatchServer(session,event.Data.(*ss.Message))
						}
					})
					break
				case *rpc.RPCRequest:
					queue.PostNoWait(func() {
						rpcServer.OnRPCMessage(rpcchan,event.Data.(*rpc.RPCRequest))
					})
					break	
				case *rpc.RPCResponse:
					conn.rpcCli.OnRPCMessage(event.Data.(*rpc.RPCResponse))
					break
				default:
					Errorf("invaild message type\n")
					break
			}
		}
	})

	addSessionPeerID(session,end.toPeerID())
	pendingMsg := end.pendingMsg
	end.pendingMsg = end.pendingMsg[0:0]
	Infof("Flush pending message:%d\n",len(pendingMsg))
	for _ , v := range pendingMsg {
		session.Send(v)
	}	
	pendingCall := end.pendingCall
	end.pendingCall = end.pendingCall[0:0]	 
		
	now := time.Now()

	for _ , v := range pendingCall {
		if now.After(v.deadline) {
			//连接消费的事件已经超过请求的超时时间
			v.cb(nil,fmt.Errorf("rpc call timeout"))
		} else {
			remain := v.deadline.Sub(now)/time.Millisecond //剩余时间(ms)
			if remain <= 1 {
				//剩余1毫秒,几乎肯定超时，没有调用的必要
				v.cb(nil,fmt.Errorf("rpc call timeout"))
			} else{
				err := conn.rpcCli.AsynCall("rpc",v.arg,uint32(remain),v.cb)//methord参数在这里没有被使用，随便填一个值
				if nil != err {
					v.cb(nil,err)
				}
			}
		}
	}
}